package cn.doit.report

import cn.doit.common.{DolphinAppComm, DolphinKPI}
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.SQLContext
import scalikejdbc.DB

object AppAnalysisCore {
  def main(args: Array[String]): Unit = {
    if (args.size != 2) {
      println(
        """
          |Usage: cn.doit.report.Save2Redis
          |	<parquetInputPath> <appdictFilePath>
          | "请输入2个参数"
        """.stripMargin)
      sys.exit(-1)
    }

    val Array(parquetInputPath, appdictFilePath) = args

    val sc = DolphinAppComm.creatSparkContext("cn.doit.report.Save2Redis")

    /**广播字典数据(全量数据)*/
    val appdictDataMap = sc.textFile(appdictFilePath)
      .map(line=>line.split("\t",-1))
      .filter(_.size>=5)
      .map(arr=>(arr(4),arr(1))).collectAsMap()

    val appdictBT = sc.broadcast(appdictDataMap)
    val sQLContext = new SQLContext(sc)
    //读取parquet文件
    val dataFrame = sQLContext.read.parquet(parquetInputPath)
    //整理数据
    dataFrame.map(row => {
      val day = row.getAs[String]("requestdate").substring(0, 10)
      var appName = row.getAs[String]("appname")
      if (StringUtils.isEmpty(appName)) {
        //如果appName为空则更具appid去字典文件中查找appName
        val appID = row.getAs[String]("appid")
        if (StringUtils.isEmpty(appID)) {
          appName = appdictBT.value.getOrElse(appID, appID)
        } else appName = "未知"
      }
      ((day, appName),DolphinKPI.caculateKpi(row))

    })
      .reduceByKey((list1,list2)=>list1.zip(list2).map(tp=> tp._1+tp._2))

      .foreachPartition(partition => {
        DB localTx { implicit session =>

          /**
            * 封装参数
            * Seq(
            * Seq(20181201,appName, rawReq,.........................) 第一行数据
            * Seq(20181201,appName, rawReq,.........................) 第二行数据
            * Seq(20181201,appName, rawReq,.........................) 第N行数据
            * )
            */
          val batchParams = partition.map {
            case (
              (day, appName),
              List(
              rawRequest, effectiveRequest, adRequest,
              isRTB, succRTB, adShow, adClick, expense, cost)
              ) => Seq(
              day, appName, rawRequest, effectiveRequest, adRequest,
              isRTB, succRTB, adShow, adClick, expense, cost
            )
          }.toSeq
          sql"insert into r_app_analysis values (?, ?, ?, ?, ?, ?, ?, ?,?,?, ?)".batch(batchParams: _*).apply()
        }
      })
    sc.stop()
  }
}
